package org.shj.spark.core

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.shj.spark.util.Util

object TopNGroupDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf
    conf.setAppName("TopNGroupDemo").setMaster("local")
    val sc = new SparkContext(conf)
    
    val lines = sc.textFile(Util.fullPath("topNGrp.txt"))
    val pair = lines.map(line => {
      val arr = line.split("\\s+")
      (arr(0), (arr(2).toInt, line))
    })
    val grp = pair.groupByKey()
    val top2= grp.map(item => (item._1, item._2.toSeq.sortWith(_._1 > _._1).take(2)))
    top2.sortByKey().foreach(item => item._2.foreach(t => println(t._2)))
    
    sc.stop()
  }
}